package cn.tiansu.eway.kylinengine.service;

import java.io.File;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.log4j.Logger;
import org.springframework.jdbc.core.JdbcTemplate;

import cn.tiansu.eway.common.utils.CmdUtils;
import cn.tiansu.eway.common.utils.DBConnection;
import cn.tiansu.eway.common.utils.PropertyConfigurator;
import cn.tiansu.eway.kylinengine.manager.KYLINEngineListener;
import cn.tiansu.eway.kylinengine.model.Condition;
import cn.tiansu.eway.kylinengine.model.Dimention;
import cn.tiansu.eway.kylinengine.model.KylinResult;
import cn.tiansu.eway.kylinengine.model.Measurement;
import cn.tiansu.eway.kylinengine.model.Olaps;
import cn.tiansu.eway.kylinengine.model.RequestObject;
import cn.tiansu.eway.kylinengine.model.Results;
import cn.tiansu.eway.kylinengine.model.UpdateTenantInfo;
import cn.tiansu.eway.kylinengine.utils.DataTimeUtils;
import cn.tiansu.eway.kylinengine.utils.JsonTools;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

/**
 * olap和kylin的交互模块
 * 
 * @Title: KYLINEngineService.java
 * @author: 吴鹏
 * @date: 2015年8月25日
 *
 */
public class KYLINEngineService
{
	/**
	 * 实例
	 */
	private static KYLINEngineService instance = new KYLINEngineService();

	/**
	 * 调用kylin API的URI
	 */
	private String PREFIX_URL = "http://"
			+ PropertyConfigurator.getInstance(null, "config.properties")
					.getParameter("kylin_url") + "/kylin/api/";

	/**
	 * 所有的维度表以及事实表 常量
	 */
	private static String TABLE_FACT = "fact_samples";

	private static String TABLE_DIM_CATEGORY = "dim_category";

	private static String TABLE_DIM_REGION = "dim_region";

	private static String TABLE_DIM_TIME = "dim_time";

	private static String TABLE_DIM_BUS1 = "dim_business1";

	private static String TABLE_DIM_BUS2 = "dim_business2";

	private static String TABLE_DIM_BUS3 = "dim_business3";

	private static String alias = "mysql";

	/**
	 * 维度表集合
	 */
	// private static String[] tableArr =
	// { TABLE_DIM_CATEGORY, TABLE_DIM_REGION, TABLE_DIM_TIME, TABLE_DIM_BUS1,
	// TABLE_DIM_BUS2, TABLE_DIM_BUS3 };

	private final static Logger log = Logger
			.getLogger(KYLINEngineService.class);

	private KYLINEngineService()
	{
	}

	/**
	 * 实例化
	 * 
	 * @return instance
	 */
	public static KYLINEngineService getInstance()
	{
		return instance;
	}

	public void updateTenant(String message)
	{
		JSONObject jsonObject = JSONObject.parseObject(message);
		JSONObject bodyObj = jsonObject.getJSONObject("body");
		UpdateTenantInfo info = JsonTools.jsonToObject(
				bodyObj.getString("msg"), UpdateTenantInfo.class);

		if (info == null)
		{
			return;
		}

		addTenant(info.getAddTenant());
		delTenant(info.getDelTenant());
	}

	/**
	 * 添加租户
	 * 
	 * @param message
	 *            消息
	 */
	private void addTenant(List<String> tenantList)
	{

		log.info("add tenantList=" + tenantList);

		if (tenantList != null && !tenantList.isEmpty())
		{
			for (String tenant : tenantList)
			{
				// 调用接口命令实现以下功能：
				// 1.在 data 目录下创建以 租户名称为 目录的 子目录
				// 2.在hive中添加租户数据库
				// 3.创建事实表和维度表
				// 4.维度表初始化数据
				tenant = tenant.toUpperCase();

				if (KYLINEngineListener.tenantSet.contains(tenant))
				{
					log.info("tenant " + tenant + " has exists!");
					continue;
				}

				// 创建data目录
				File tenantDir = new File(System.getProperty("user.dir")
						+ File.separator + "data" + File.separator + tenant);

				if (!tenantDir.exists())
				{
					tenantDir.mkdir();
				} else
				{
					log.warn("tenant [" + tenant
							+ "] 's dataFile is exists,please contact admin !");
					continue;
				}

				// 创建bak目录
				File tenantBakDir = new File(System.getProperty("user.dir")
						+ File.separator + "bak" + File.separator + tenant);

				if (!tenantBakDir.exists())
				{
					tenantBakDir.mkdir();
				}

				String path = System.getProperty("user.dir") + File.separator
						+ "bin" + File.separator + "sample.sh " + tenant;

				CmdUtils.execShell(path);

				synchronized (KYLINEngineListener.tenantSet)
				{
					KYLINEngineListener.tenantSet.add(tenant);
				}

			}

			reloadMeta();
		}
	}

	/**
	 * 刪除租戶
	 * 
	 * @param tenant
	 *            租戶名称
	 */
	private void delTenant(List<String> tenantList)
	{
		log.info("del tenantList=" + tenantList);

		if (tenantList != null && !tenantList.isEmpty())
		{
			for (String tenant : tenantList)
			{
				// 暂时先不注销立方，只是在缓存中去掉租户，使之不在周期性刷新
				// disableCube(tenant);
				synchronized (KYLINEngineListener.tenantSet)
				{
					KYLINEngineListener.tenantSet.remove(tenant);
				}
			}

		}
	}

	/**
	 * 重新加载
	 */
	public void reloadMeta()
	{
		/**
		 * /cache/{type}/{name}/{action} type - required string ‘METADATA’ or
		 * ‘CUBE’ name - required string Cache key. action - required string
		 * ‘create’, ‘update’ or ‘drop’
		 **/
		String url = PREFIX_URL + "cache/all/all/update";
		String param = "{\"event\":\"all\"}";
		JSONObject jsonParam = JSONObject.parseObject(param);
		HttpClientMethod.httpPut(url, jsonParam, false);
	}

	/**
	 * 周期性的构造立方
	 */
	public void initCube(String tenant, long period)
	{
		log.info("开始刷新立方");
		log.info("开始导入 fact_samples数据，租户是 " + tenant);

		// 先导入事实表的数据到hive中
		boolean success = initData2Hive("fact_samples", tenant);

		if (!success)
		{
			log.info("本次终止构建立方，租户是 " + tenant);
			return;
		}

		// 构建立方
		log.info("开始构建立方，租户是 " + tenant);
		buildCube(tenant, period,new DataTimeUtils());
	}

	public static void main(String[] args)
	{
		
		String url = "http://"
				+ PropertyConfigurator.getInstance(null, "config.properties")
				.getParameter("kylin_url") + "/kylin/api/" + "cubes/" 
				+ "NEWCITY_fact_samples_cube/rebuild";
		DataTimeUtils dt = new DataTimeUtils();
		// 当前时间计算上一个周期的增量立方,上一个周期的结束时间为当前周期的0时或0分
		long startSec = dt.localTime2UTCTime("1970-01-01 00:00:00",
				DataTimeUtils.timeFotmat1);

		long endSec = dt.localTime2UTCTime("2016-02-25 00:00:00",
				DataTimeUtils.timeFotmat1);
		
//		System.out.println(DataTimeUtils.localTime2UTCTime("2016-02-24 16:00:00",
//				DataTimeUtils.timeFotmat1));
//		System.out.println(DataTimeUtils.localTime2UTCTime("2016-02-25",
//				DataTimeUtils.timeFotmat3));

//		String param = "{\"startTime\":" + startSec + ",\"endTime\":" + endSec
//				+ ",\"buildType\":\"MERGE\",\"forceMergeEmptySegment\":true}";
//		JSONObject jsonParam = JSONObject.parseObject(param);
//
//		HttpClientMethod.httpPut(url, jsonParam, false);
		
		System.out.println("2016-03-09");
		System.out.println(dt.localTime2UTCTime(dt.getDayDate(),
				DataTimeUtils.timeFotmat3));
		
	}

	/**
	 * 将立方置为不可用
	 * 
	 * @param tenant
	 *            租户名称
	 */
	private void disableCube(String tenant)
	{
		String url = PREFIX_URL + "cubes/" + tenant.toUpperCase()
				+ "_fact_samples_cube/disable";

		String param = "{}";
		JSONObject jsonParam = JSONObject.parseObject(param);

		HttpClientMethod.httpPut(url, jsonParam, false);

	}

	// private static void dropCube(String tenant)
	// {
	// String url = PREFIX_URL + "cubes/" + tenant + "/drop";
	//
	// String param = "{}";
	// JSONObject jsonParam = JSONObject.parseObject(param);
	//
	// JSONObject obj = HttpClientMethod.httpPut(url, jsonParam, false);
	// }

	/**
	 * 判断数据文件和维度文件是否存在，是否有数据
	 * 
	 * @param tenant
	 *            租户名称
	 * @return 如果维度文件有一个不存在，返回false;如果地域，分类分项维度为空，返回false
	 */
	private boolean kylinDataFileisExist(String tenant)
	{
		String path = System.getProperty("user.dir") + File.separator + "data"
				+ File.separator + tenant + File.separator;

		File dataFile = new File(path + "fact_samples.txt");
		if (!dataFile.exists() || dataFile.length() == 0)
		{
			log.warn("file " + path + "fact_samples.txt"
					+ " is not exist or is empty !!!");
			return true;
		}

		File cateFile = new File(path + "vw_dim_category.txt");
		if (!cateFile.exists() || cateFile.length() == 0)
		{
			log.warn("file " + path + "vw_dim_category.txt"
					+ " is not exist or is empty,kylin is stop !!!");
			return false;
		}

		File regionFile = new File(path + "vw_dim_region.txt");
		if (!regionFile.exists() || regionFile.length() == 0)
		{
			log.warn("file " + path + "vw_dim_region.txt"
					+ " is not exist or is empty,kylin is stop !!!");
			return false;
		}

		if (!new File(path + "dim_business1.txt").exists())
		{
			log.warn("file " + path + "dim_business1.txt"
					+ " is not exist,kylin is stop !!!");
			return true;
		}

		if (!new File(path + "dim_business2.txt").exists())
		{
			log.warn("file " + path + "dim_business2.txt"
					+ " is not exist,kylin is stop !!!");
			return true;
		}

		if (!new File(path + "dim_business3.txt").exists())
		{
			log.warn("file " + path + "dim_business3.txt"
					+ " is not exist,kylin is stop !!!");
			return true;
		}

		return true;
	}

	/**
	 * 把数据导入到hive中
	 * 
	 * @param tables
	 *            需要导入的表，database.tablename
	 * @param tenant
	 *            租户
	 * @return
	 */
	private boolean initData2Hive(String tables, String tenant)
	{
		// 如果维度文件有一个不存在，终止;如果地域，分类分项维度为空有一个为空，终止
		if (!kylinDataFileisExist(tenant))
		{
			return false;
		}

		// 将新的fact数据导入到hive中
		String path = System.getProperty("user.dir") + File.separator + "bin"
				+ File.separator + "factDataToHive.sh " + tenant;

		CmdUtils.execShell(path);

		return true;

	}

	/**
	 * 调用kylin接口构造立方
	 */
	private void buildCube(String tenant, long period,DataTimeUtils dt)
	{
		String url = PREFIX_URL + "cubes/" + tenant.toUpperCase()
				+ "_fact_samples_cube/rebuild";

		log.info("buildCube  url is " + url);

		// 当前时间计算上一个周期的增量立方,上一个周期的结束时间为当前周期的0时或0分
		long endSec = dt.localTime2UTCTime(dt.getDayDate(),
				DataTimeUtils.timeFotmat3);
		
		long startSec = endSec - 1000l * period;

		String param = "{\"startTime\":" + startSec + ",\"endTime\":" + endSec
				+ ",\"buildType\":\"BUILD\"}";
		JSONObject jsonParam = JSONObject.parseObject(param);

		HttpClientMethod.httpPut(url, jsonParam, false);
	}
	
	
	/**
	 * 
	 * @param tenant
	 *            租户
	 * @param startTime
	 *            开始时间
	 * @param endTime
	 *            结束时间
	 */
	public void refreshCubeByTime(String tenant, String startTime,
			String endTime)
	{
		String url = PREFIX_URL + "cubes/" + tenant.toUpperCase()
				+ "_fact_samples_cube/rebuild";
		
		DataTimeUtils dt = new DataTimeUtils();
		
		// 当前时间计算上一个周期的增量立方,上一个周期的结束时间为当前周期的0时或0分
		long startSec = dt.localTime2UTCTime(startTime,
				DataTimeUtils.timeFotmat3);

		long endSec = dt.localTime2UTCTime(endTime,
				DataTimeUtils.timeFotmat3);

		String param = "{\"startTime\":" + startSec + ",\"endTime\":" + endSec
				+ ",\"buildType\":\"REFRESH\"}";
		JSONObject jsonParam = JSONObject.parseObject(param);

		HttpClientMethod.httpPut(url, jsonParam, false);
	}
	
	public void mergeCube(String tenant, String startTime,
			String endTime,DataTimeUtils dt)
	{
		String url = PREFIX_URL + "cubes/" + tenant.toUpperCase()
				+ "_fact_samples_cube/rebuild";

		// 当前时间计算上一个周期的增量立方,上一个周期的结束时间为当前周期的0时或0分
		long startSec = dt.localTime2UTCTime(startTime,
				DataTimeUtils.timeFotmat1);

		long endSec = dt.localTime2UTCTime(endTime,
				DataTimeUtils.timeFotmat1);

		String param = "{\"startTime\":" + startSec + ",\"endTime\":" + endSec
				+ ",\"buildType\":\"MERGE\",\"forceMergeEmptySegment\":true}";
		JSONObject jsonParam = JSONObject.parseObject(param);

		HttpClientMethod.httpPut(url, jsonParam, false);
	}

	/**
	 * 根据请求获取数据
	 * 
	 * @param message
	 *            请求信息
	 * @return 数据json对象
	 */
	public JSONObject queryDataByCondition(String message)
	{
		JSONObject jsonObject = JSONObject.parseObject(message);
		String bodyStr = jsonObject.getString("body");

		log.debug("bodyStr  = " + bodyStr);

		// 将请求json字符串转化为 对象
		RequestObject obj = JsonTools
				.jsonToObject(bodyStr, RequestObject.class);

		if (obj.getOlaps() == null || obj.getOlaps().size() == 0)
		{
			return null;
		}

		// 根据租户名获取 hive的数据库分库
		String tenant = obj.getOlaps().get(0).getTenant();
		String url = PREFIX_URL + "query";

		// 构造查询sql
		String sql = initSql(obj.getOlaps().get(0));

		// 构造符合kylin API的的请求参数
		String param = "{\"sql\":\""
				+ sql
				+ "\",\"offset\":0,\"limit\":1000,\"acceptPartial\":true,\"project\":\""
				+ tenant + "\"}";
		JSONObject jsonParam = JSONObject.parseObject(param);

		// 调用kylin api请求 查询数据
		JSONObject jsonObj = HttpClientMethod.httpPost(url, jsonParam, false);

		if (jsonObj == null)
		{
			return new JSONObject();
		}

		// 将kylin返回数据构造成 对象
		KylinResult res = JsonTools.jsonToObject(jsonObj.toJSONString(),
				KylinResult.class);

		log.info("****************** " + res.getResults());
		log.info("****************** " + res.getColumnMetas());

		// 将kylin返回数据 封装成
		return dealResult(res.getColumnMetas(), res.getResults());
	}

	/**
	 * 对kylin结果进行处理
	 * 
	 * @param columnMetas
	 *            列信息meta
	 * @param results
	 *            kylin查询结果
	 * @return 结果json对象
	 */
	private JSONObject dealResult(List<Map<String, String>> columnMetas,
			List<String> results)
	{
		List<String> dimentions = new ArrayList<String>();
		List<String> datas = new ArrayList<String>();

		for (Map<String, String> map : columnMetas)
		{
			dimentions.add(map.get("label"));
		}

		for (String str : results)
		{
			datas.add(str.replace("\"", ""));
		}

		Map<String, List<String>> map = new HashMap<String, List<String>>();
		map.put("dimentions", dimentions);
		map.put("datas", datas);

		Results r = new Results();
		List<Map<String, List<String>>> resultList = new ArrayList<Map<String, List<String>>>();
		resultList.add(map);
		r.setResults(resultList);

		log.info("###############   " + JSON.parseObject(JSON.toJSONString(r)));
		return JSON.parseObject(JSON.toJSONString(r));
	}

	/**
	 * 构造查询sql
	 * 
	 * @param olap
	 *            请求对象
	 * @return sql
	 */
	private String initSql(Olaps olap)
	{
		StringBuilder sb = new StringBuilder("SELECT ");

		StringBuilder measurementsSB = new StringBuilder();
		Set<String> tableSet = new HashSet<String>();
		StringBuilder conditionSB = new StringBuilder();
		StringBuilder groupBySB = new StringBuilder();

		List<Measurement> measurements = olap.getMeasurements();
		List<Dimention> dimentions = olap.getDimentions();
		List<Condition> conditions = olap.getConditions();

		// 构造聚合 算法表达式
		if (measurements != null && !measurements.isEmpty())
		{
			for (int i = 0; i < measurements.size(); i++)
			{
				String name = measurements.get(i).getName();

				if (i != 0)
				{
					measurementsSB.append(",");
				}
				measurementsSB.append(name).append("(sample_value) as ")
						.append("nts_").append(name);
			}
		}
		
		//为了对已修复数据增加标记位，需要对聚合查询中统计数据进行标记，所以每次查询带上标记位
		measurementsSB.append(",max(status) as status");

		// 构造 where 语句
		if (conditions != null && !conditions.isEmpty())
		{
			for (int i = 0; i < conditions.size(); i++)
			{
				List<Map<String, String>> levels = conditions.get(i)
						.getLevels();
				String dim = conditions.get(i).getDim();

				// 基于condition的特殊性，list长度必然为1
				Map<String, String> conMap = levels.get(0);

				Iterator<Map.Entry<String, String>> it = conMap.entrySet()
						.iterator();

				while (it.hasNext())
				{
					Entry<String, String> en = it.next();

					if (conditionSB.length() > 0)
					{
						conditionSB.append(" and ");
					}

					if (TABLE_DIM_TIME.equals(dim))
					{
						conditionSB.append(dim).append(".").append(en.getKey())
								.append("=").append(en.getValue());
					} else
					{
						conditionSB.append(dim).append(".").append(en.getKey())
								.append("='").append(en.getValue()).append("'");
					}
				}

				tableSet.add(dim);
			}
		}

		// 构造 维度，由维度 会 衍生出 condition语句和 group by 语句 和 select 语句
		StringBuilder dimConSql = new StringBuilder();
		if (dimentions != null && !dimentions.isEmpty())
		{
			for (int i = 0; i < dimentions.size(); i++)
			{
				List<Map<String, String>> levels = dimentions.get(i)
						.getLevels();
				String dim = dimentions.get(i).getDim();

				if (i > 0)
				{
					dimConSql.append(" and ");
				}

				dimConSql.append("(");

				// dimentions对于同一维度可能会有多个值，例如 {"dim":
				// "dim_category","levels":[{"category_Level1":
				// "能耗","category_Level2": "水"},{"category_Level1":
				// "能耗","category_Level2": "电"}]
				// 这种情况在select 只需要 对其中一个解析 就行，多个解析存在冗余，但在condition需要解析多个值，并且每个值
				// 之间是 or的关系

				for (int len = 0; len < levels.size(); len++)
				{
					if (len > 0)
					{
						dimConSql.append(" or ");
					}
					dimConSql.append("(");
					Map<String, String> level = levels.get(len);

					Iterator<Map.Entry<String, String>> it = level.entrySet()
							.iterator();

					// int mapSize = 0;
					dimConSql.append(" 1=1 ");

					while (it.hasNext())
					{
						Entry<String, String> en = it.next();
						String key = en.getKey();
						String value = en.getValue();

						if (len == 0)
						{
							// select 中加载
							if (measurementsSB.length() > 1)
							{
								measurementsSB.append(",");
							}
							measurementsSB.append(key);

							// group by 加载
							if (groupBySB.length() > 1)
							{
								groupBySB.append(",");
							}
							groupBySB.append(key);
						}

						// where中加载
						if (!value.equals("children"))
						{

							// if (mapSize > 0)
							// {
							dimConSql.append(" and ");
							// }

							if (TABLE_DIM_TIME.equals(dim))
							{
								dimConSql.append(dim).append(".").append(key)
										.append("=").append(value);
							} else
							{
								dimConSql.append(dim).append(".").append(key)
										.append("='").append(value).append("'");
							}

							// mapSize++;
						}

					}

					dimConSql.append(")");
				}

				dimConSql.append(")");

				tableSet.add(dim);
			}
		}

		if (conditionSB.toString().equals(""))
		{
			conditionSB.append(dimConSql);
		} else
		{
			conditionSB.append(" and ").append(dimConSql);
		}

		// 构造 关联表 语句
		StringBuilder tableSql = new StringBuilder();
		tableSql.append(TABLE_FACT);

		if (tableSet.contains(TABLE_DIM_REGION))
		{
			tableSql.append(preTableSql(TABLE_DIM_REGION, "region_id"));
		}
		if (tableSet.contains(TABLE_DIM_TIME))
		{
			tableSql.append(preTableSql(TABLE_DIM_TIME, "time_id"));
		}
		if (tableSet.contains(TABLE_DIM_CATEGORY))
		{
			tableSql.append(preTableSql(TABLE_DIM_CATEGORY, "category_id"));
		}
		if (tableSet.contains(TABLE_DIM_BUS1))
		{
			tableSql.append(preTableSql(TABLE_DIM_BUS1, "business1_id",
					"business_id"));
		}
		if (tableSet.contains(TABLE_DIM_BUS2))
		{
			tableSql.append(preTableSql(TABLE_DIM_BUS2, "business2_id",
					"business_id"));
		}
		if (tableSet.contains(TABLE_DIM_BUS3))
		{
			tableSql.append(preTableSql(TABLE_DIM_BUS3, "business3_id",
					"business_id"));
		}

		sb.append(measurementsSB).append(" from ").append(tableSql)
				.append(" where ").append(conditionSB).append(" group by ")
				.append(groupBySB);

		log.info("select sql : " + sb.toString());
		return sb.toString();
	}

	/**
	 * 构造表&表的关联sql
	 * 
	 * @param table
	 *            维度表
	 * @param column
	 *            关联列
	 * @return sql
	 */
	private StringBuilder preTableSql(String table, String column)
	{
		StringBuilder sql = new StringBuilder();
		if (table.indexOf("business") > -1)
		{
			sql.append(" left join ");
		} else
		{
			sql.append(" inner join ");
		}

		sql.append(table).append(" on ").append(TABLE_FACT).append(".")
				.append(column).append("=").append(table).append(".")
				.append(column);
		return sql;
	}

	/**
	 * 构造表&表的关联sql (事实表和维度表的列名不一致)
	 * 
	 * @param table
	 *            维度表
	 * @param factColumn
	 *            事实表的关联列
	 * @param lookupColumn
	 *            维度表的关联列
	 * @return sql
	 */
	private StringBuilder preTableSql(String table, String factColumn,
			String lookupColumn)
	{
		StringBuilder sql = new StringBuilder();

		if (table.indexOf("business") > -1)
		{
			sql.append(" left join ");
		} else
		{
			sql.append(" inner join ");
		}
		sql.append(table).append(" on ").append(TABLE_FACT).append(".")
				.append(factColumn).append("=").append(table).append(".")
				.append(lookupColumn);
		return sql;
	}

	public Set<String> getTenantName()
	{
		String prefix = PropertyConfigurator.getInstance(null,
				"config.properties").getParameter("prefix");

		String sqlTenant = "select name from " + prefix
				+ "_cloud.tb_tenant where enabled=1";

		JdbcTemplate jt = DBConnection.getInstance().getJdbcTemplate(alias);

		List<Map<String, String>> list = jt.queryForList(sqlTenant);

		Set<String> tenantSet = new HashSet<String>();
		for (Map<String, String> map : list)
		{
			tenantSet.add(map.get("name").toUpperCase());
		}

		return tenantSet;
	}
}
